package com.leilei;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.core.util.StrUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.Random;
import java.util.stream.StreamSupport;

/**
 * @author lei
 * @version 1.0
 * @date 2021-04-25 12:29
 * @desc
 */
public class AA {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(12);
        DataStreamSource<Integer> elements = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10,11,12);
        final SingleOutputStreamOperator<String> operator = elements.keyBy(e -> e % 2 == 0)
                .countWindow(2)
                .apply(new Mywindow());
        operator.print();
        env.execute();
    }

    public static class Mywindow extends RichWindowFunction<Integer, String, Boolean, GlobalWindow> {
        MapState<String, Integer> mapState;

        @Override
        public void apply(Boolean aBoolean, GlobalWindow window, Iterable<Integer> input, Collector<String> out) throws Exception {
            System.out.println("key:" + aBoolean);
            StreamSupport.stream(input.spliterator(), true).map(String::valueOf).forEach(out::collect);
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            MapStateDescriptor<String, Integer> zs = new MapStateDescriptor<>("zs",
                    TypeInformation.of(String.class),
                    TypeInformation.of(Integer.class));
            mapState = getRuntimeContext().getMapState(zs);
            System.out.println("窗口打开");
        }

    }

}
